package socket

import (
	"context"
	"device-modbus/internal/biz/dal/dchan"
	"device-modbus/internal/biz/dao"
	"device-modbus/internal/biz/model"
	"device-modbus/internal/biz/query"
	"device-modbus/internal/biz/repo"
	"device-modbus/internal/logic"
	"device-modbus/internal/plugin_manager"
	"device-modbus/internal/transpond"
	"device-modbus/plugin"
	"encoding/hex"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/cloudwego/hertz/pkg/common/hlog"
	"github.com/panjf2000/ants/v2"
	"github.com/panjf2000/gnet/v2"
	"golang.org/x/sync/errgroup"
)

// Config 服务器配置参数结构体
type Config struct {
	ListenAddr             string
	AuthTimeout            time.Duration
	PingInterval           time.Duration // 增加默认值，避免过短
	MaxMissedPings         int
	PingResponseTimeout    time.Duration
	WorkerPoolSize         int
	SendChanBufferSize     int           // 发送通道缓冲区大小
	StateSyncBatchSize     int           // 状态同步批量处理大小
	StateSyncQueryBatch    int           // 状态同步查询批次大小
	StateSyncDelay         time.Duration // 启动后延迟多久开始同步状态
	StateSyncInterval      time.Duration // 状态同步间隔
	NotificationBufferSize int           // 通知队列缓冲区大小
	ShutdownTimeout        time.Duration // 优雅关闭超时时间
	MaxConnections         int32         // 最大连接数，改为可配置
}

// DefaultConfig 返回默认配置 - 调整默认值以降低CPU占用
func DefaultConfig() Config {
	return Config{
		ListenAddr:             "tcp://0.0.0.0:9000",
		AuthTimeout:            60 * time.Second,
		PingInterval:           120 * time.Second, // 延长心跳间隔
		MaxMissedPings:         3,
		PingResponseTimeout:    60 * time.Second,
		WorkerPoolSize:         20, // 减少工作池大小
		SendChanBufferSize:     100,
		StateSyncBatchSize:     200, // 增大批次减少循环次数
		StateSyncQueryBatch:    200,
		StateSyncDelay:         5 * time.Minute,  // 延长启动后同步延迟
		StateSyncInterval:      30 * time.Minute, // 状态同步间隔
		NotificationBufferSize: 100,
		ShutdownTimeout:        30 * time.Second,
		MaxConnections:         5000, // 合理设置最大连接数
	}
}

// DeviceOnlineNotification 设备上线通知结构体
type DeviceOnlineNotification struct {
	DeviceID     string    // 设备ID
	Iccid        string    // 设备ICCID
	RemoteAddr   string    // 远程连接地址
	OnlineTime   time.Time // 上线时间
	LastOffline  time.Time // 上次离线时间
	IsReconnect  bool      // 是否是重连
	OnlineStatus string    // 上线状态 0 离线 1在线
}

// DeviceBizNotification 设备业务通知结构体
type DeviceBizNotification struct {
	DeviceID   string      // 设备ID
	Iccid      string      // 设备ICCID
	BizTime    int64       // 业务时间
	BatchTime  int64       // 批次时间 时间戳（秒）
	BizType    string      // 业务类型
	Data       interface{} // 业务数据
	SensorAddr string      // 传感器地址
}

/* ---------- 连接管理 ---------- */
type connManager struct {
	temp             sync.Map          // 未认证连接: remoteAddr -> *Client
	auth             sync.Map          // 已认证连接: remoteAddr -> *Client
	banned           sync.Map          // 被封禁IP: remoteAddr -> time.Time (解封时间)
	mu               sync.RWMutex      // 读写锁分离，提高并发性能
	deviceMap        map[string]string // 设备ID到remoteAddr的映射
	shuttingDown     bool              // 是否正在关闭
	totalConnections int32             // 总连接数计数
	maxConnections   int32             // 最大允许连接数
}

// 初始化连接管理器时使用配置的最大连接数
func newConnManager(maxConnections int32) *connManager {
	return &connManager{
		deviceMap:      make(map[string]string),
		maxConnections: maxConnections,
	}
}

// BanIP 封禁指定IP一段时间
func (m *connManager) BanIP(addr string, duration time.Duration) {
	expiration := time.Now().Add(duration)
	m.banned.Store(addr, expiration)
	hlog.Infof("IP %s 已被封禁 %v，到期时间: %v", addr, duration, expiration)
}

// IsIPBanned 检查IP是否处于封禁状态
func (m *connManager) IsIPBanned(addr string) bool {
	v, ok := m.banned.Load(addr)
	if !ok {
		return false
	}

	expiration, ok := v.(time.Time)
	if !ok {
		m.banned.Delete(addr) // 清理无效记录
		return false
	}

	if time.Now().After(expiration) {
		m.banned.Delete(addr) // 自动解封过期IP
		return false
	}

	return true
}

func (m *connManager) moveToAuth(c *Client) {
	if m.shuttingDown {
		return
	}

	m.temp.Delete(c.GetRemoteAddr())
	m.auth.Store(c.GetRemoteAddr(), c)

	m.mu.Lock()
	m.deviceMap[c.GetDeviceID()] = c.GetRemoteAddr()
	m.mu.Unlock()

	c.SetAuthenticated(true)
}

func (m *connManager) removeAuth(c *Client) {
	m.auth.Delete(c.GetRemoteAddr())

	m.mu.Lock()
	delete(m.deviceMap, c.GetDeviceID())
	m.mu.Unlock()
}

func (m *connManager) getClientByDeviceID(deviceID string) (*Client, bool) {
	m.mu.RLock()
	addr, ok := m.deviceMap[deviceID]
	m.mu.RUnlock()

	if !ok {
		return nil, false
	}

	v, ok := m.auth.Load(addr)
	if !ok {
		return nil, false
	}

	client, ok := v.(*Client)
	return client, ok
}

// 获取所有已认证客户端，使用Range避免大范围锁定
func (m *connManager) getAllAuthClients() []*Client {
	var clients []*Client
	m.auth.Range(func(key, value interface{}) bool {
		if client, ok := value.(*Client); ok {
			clients = append(clients, client)
		}
		return true
	})
	return clients
}

func (m *connManager) getAllTempClients() []*Client {
	var clients []*Client
	m.temp.Range(func(key, value interface{}) bool {
		if client, ok := value.(*Client); ok {
			clients = append(clients, client)
		}
		return true
	})
	return clients
}

func (m *connManager) setShuttingDown() {
	m.shuttingDown = true
}

/* ---------- 设备状态同步器 ---------- */
type DeviceStateSyncer struct {
	cfg            Config
	pool           *ants.Pool
	server         *TcpServer
	startTime      time.Time
	ctx            context.Context
	cancel         context.CancelFunc
	totalProcessed int64        // 总处理设备数
	totalOffline   int64        // 总离线设备数
	syncTicker     *time.Ticker // 定期同步的定时器
}

func NewDeviceStateSyncer(server *TcpServer, pool *ants.Pool, cfg Config) *DeviceStateSyncer {
	ctx, cancel := context.WithCancel(context.Background())
	syncer := &DeviceStateSyncer{
		cfg:       cfg,
		pool:      pool,
		server:    server,
		startTime: time.Now(),
		ctx:       ctx,
		cancel:    cancel,
	}

	// 设置定期同步定时器
	if cfg.StateSyncInterval > 0 {
		syncer.syncTicker = time.NewTicker(cfg.StateSyncInterval)
		go syncer.startPeriodicSync()
	}

	return syncer
}

// 定期同步设备状态
func (s *DeviceStateSyncer) startPeriodicSync() {
	defer s.syncTicker.Stop()

	for {
		select {
		case <-s.ctx.Done():
			return
		case <-s.syncTicker.C:
			hlog.Info("开始定期同步设备状态...")
			ctx, cancel := context.WithTimeout(s.ctx, 10*time.Minute)
			if err := s.SyncAllDevices(ctx); err != nil {
				hlog.Errorf("定期设备状态同步失败: %v", err)
			}
			cancel()
		}
	}
}

func (s *DeviceStateSyncer) Stop() {
	s.cancel()
}

// SyncAllDevices 分批次查询并同步设备状态 - 优化：单线程处理，减少无用循环
func (s *DeviceStateSyncer) SyncAllDevices(ctx context.Context) error {
	hlog.Info("开始同步设备状态...")

	// 重置统计信息
	s.totalProcessed = 0
	s.totalOffline = 0

	// 获取总在线设备数（用于进度跟踪）
	totalOnline, err := s.getTotalOnlineDevices(ctx)
	if err != nil {
		return fmt.Errorf("获取在线设备总数失败: %w", err)
	}
	hlog.Infof("系统中标记为在线的设备总数: %d，将分批次验证状态", totalOnline)

	if totalOnline == 0 {
		hlog.Info("没有需要验证的在线设备，同步结束")
		return nil
	}

	// 优化：开机同步仅使用1个线程
	workerCount := 1

	// 计算进度报告间隔，避免频繁日志输出
	progressInterval := totalOnline / 5 // 每20%报告一次
	if progressInterval == 0 {
		progressInterval = 1
	}

	deviceChan := make(chan *model.Device, workerCount*2)
	resultChan := make(chan string, s.cfg.StateSyncBatchSize)
	errChan := make(chan error, workerCount)

	var wg sync.WaitGroup

	// 启动单工作线程
	wg.Add(workerCount)
	for i := 0; i < workerCount; i++ {
		go func(workerID int) {
			defer wg.Done()
			s.processDeviceWorker(ctx, workerID, deviceChan, resultChan, errChan)
		}(i)
	}

	// 启动分批次发送设备到工作池的协程
	sendDone := make(chan struct{})
	go func() {
		defer func() {
			close(deviceChan)
			close(sendDone)
		}()

		offset := 0
		for {
			select {
			case <-ctx.Done():
				hlog.Warnf("上下文已取消，停止发送设备到工作池，已处理 %d/%d", s.totalProcessed, totalOnline)
				return
			default:
				batch, err := s.queryOnlineDevicesBatch(ctx, offset, s.cfg.StateSyncQueryBatch)
				if err != nil {
					hlog.Errorf("查询设备批次失败(offset: %d): %v", offset, err)
					errChan <- fmt.Errorf("查询设备批次失败: %w", err)
					return
				}
				if len(batch) == 0 {
					hlog.Infof("所有设备批次已发送完成，共发送 %d 个设备", s.totalProcessed)
					return
				}

				// 发送当前批次设备到工作池
				for _, device := range batch {
					select {
					case <-ctx.Done():
						hlog.Warnf("上下文已取消，批次发送中断，已发送 %d/%d", s.totalProcessed, totalOnline)
						return
					case deviceChan <- device:
						s.totalProcessed++
						// 减少日志输出频率
						if s.totalProcessed%progressInterval == 0 || s.totalProcessed == totalOnline {
							progress := float64(s.totalProcessed) / float64(totalOnline) * 100
							hlog.Infof("设备状态验证进度: %.1f%% (%d/%d)", progress, s.totalProcessed, totalOnline)
						}
					}
				}

				offset += len(batch)
				// 查询间隔，避免频繁访问数据库
				time.Sleep(100 * time.Millisecond)
			}
		}
	}()

	// 等待工作池完成并关闭结果通道
	go func() {
		wg.Wait()
		<-sendDone
		close(resultChan)
		close(errChan)
		hlog.Info("所有设备状态验证完成")
	}()

	// 处理验证结果和错误
	return s.processResults(ctx, resultChan, errChan, totalOnline)
}

// getTotalOnlineDevices 获取标记为在线的设备总数
func (s *DeviceStateSyncer) getTotalOnlineDevices(ctx context.Context) (int64, error) {
	deviceQuery := query.Device
	return deviceQuery.WithContext(ctx).
		Where(deviceQuery.Online.Eq(1)).
		Where(deviceQuery.LastTime.Lt(s.startTime)).
		Count()
}

// queryOnlineDevicesBatch 分批次查询在线设备
func (s *DeviceStateSyncer) queryOnlineDevicesBatch(ctx context.Context, offset int, limit int) ([]*model.Device, error) {
	deviceQuery := query.Device
	return deviceQuery.WithContext(ctx).
		Where(deviceQuery.Online.Eq(1)).
		Where(deviceQuery.LastTime.Lt(s.startTime)).
		Offset(offset).
		Limit(limit).
		Find()
}

// 工作池处理函数：验证设备是否在线
func (s *DeviceStateSyncer) processDeviceWorker(
	ctx context.Context,
	workerID int,
	deviceChan <-chan *model.Device,
	resultChan chan<- string,
	errChan chan<- error,
) {
	for device := range deviceChan {
		select {
		case <-ctx.Done():
			hlog.Warnf("worker %d: 上下文已取消，停止处理设备", workerID)
			return
		default:
			isOnline, err := s.checkDeviceOnline(device)
			if err != nil {
				errChan <- fmt.Errorf("worker %d 验证设备 %s 状态失败: %w", workerID, device.ID, err)
				continue
			}

			if !isOnline {
				hlog.Debugf("worker %d: 设备 %s 实际已离线", workerID, device.ID)
				resultChan <- device.ID
				atomic.AddInt64(&s.totalOffline, 1) // 使用原子操作计数
			} else {
				hlog.Debugf("worker %d: 设备 %s 在线状态验证通过", workerID, device.ID)
			}
		}
	}
	hlog.Infof("worker %d 完成所有设备处理", workerID)
}

// 处理验证结果：批量更新离线设备
func (s *DeviceStateSyncer) processResults(
	ctx context.Context,
	resultChan <-chan string,
	errChan <-chan error,
	totalOnline int64,
) error {
	var (
		offlineDeviceIDs []string
		batchSize        = s.cfg.StateSyncBatchSize
		errors           []error
	)

	// 同时监听结果、错误和上下文取消
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case err, ok := <-errChan:
			if !ok {
				errChan = nil
				break
			}
			errors = append(errors, err)
			hlog.Error(err.Error())
		case deviceID, ok := <-resultChan:
			if !ok {
				resultChan = nil
				break
			}
			offlineDeviceIDs = append(offlineDeviceIDs, deviceID)

			// 达到批次大小则批量更新
			if len(offlineDeviceIDs) >= batchSize {
				if err := s.batchUpdateOffline(ctx, offlineDeviceIDs); err != nil {
					errMsg := fmt.Errorf("批量更新设备离线状态失败: %w", err)
					errors = append(errors, errMsg)
					hlog.Error(errMsg.Error())
				} else {
					hlog.Infof("已批量更新 %d 个离线设备状态", len(offlineDeviceIDs))
				}
				offlineDeviceIDs = offlineDeviceIDs[:0]
			}
		}

		if resultChan == nil && errChan == nil {
			break
		}
	}

	// 处理剩余设备
	if len(offlineDeviceIDs) > 0 {
		if err := s.batchUpdateOffline(ctx, offlineDeviceIDs); err != nil {
			errMsg := fmt.Errorf("批量更新剩余 %d 个设备离线状态失败: %w", len(offlineDeviceIDs), err)
			errors = append(errors, errMsg)
			hlog.Error(errMsg.Error())
		} else {
			hlog.Infof("已更新剩余 %d 个离线设备状态", len(offlineDeviceIDs))
		}
	}

	// 输出同步统计信息
	duration := time.Since(s.startTime)
	hlog.Infof("设备状态同步统计: 总验证设备数=%d, 发现离线设备数=%d, 耗时=%v, 平均速度=%.2f个/秒",
		s.totalProcessed, s.totalOffline, duration,
		float64(s.totalProcessed)/duration.Seconds())

	if len(errors) > 0 {
		return fmt.Errorf("设备状态同步完成，但存在 %d 个错误", len(errors))
	}

	hlog.Info("设备状态同步完成")
	return nil
}

// 检查设备是否在线
func (s *DeviceStateSyncer) checkDeviceOnline(device *model.Device) (bool, error) {
	client, exists := s.server.manager.getClientByDeviceID(device.ID)
	return exists && client.IsAuthenticated(), nil
}

// 批量更新设备为离线状态
func (s *DeviceStateSyncer) batchUpdateOffline(ctx context.Context, deviceIDs []string) error {
	if len(deviceIDs) == 0 {
		return nil
	}

	deviceQuery := query.Device
	result, err := deviceQuery.WithContext(ctx).
		Where(deviceQuery.ID.In(deviceIDs...)).
		Updates(map[string]interface{}{
			"online":         0,
			"last_time":      time.Now(),
			"offline_reason": "server_restart",
		})

	if err != nil {
		return err
	}

	hlog.Infof("批量更新 %d 个设备为离线状态，实际影响 %d 个",
		len(deviceIDs), result.RowsAffected)

	deviceLogic := logic.NewDeviceLogic(ctx)
	// 使用工作池处理通知，避免阻塞
	for _, id := range deviceIDs {
		id := id // 避免闭包陷阱
		if err := s.pool.Submit(func() {
			device, err := deviceLogic.FindById(id)
			if err != nil {
				hlog.Errorf("查询设备 %s 信息失败: %v", id, err)
				return
			}
			if device == nil {
				hlog.Warnf("设备 %s 不存在，无法发送离线通知", id)
				return
			}
			s.server.sendOfflineNotification(device.ID, device.Iccid)
		}); err != nil {
			hlog.Errorf("提交设备 %s 离线通知任务失败: %v", id, err)
		}
	}

	return nil
}

// 下线通知
func (s *TcpServer) sendOfflineNotification(deviceID string, iccid string) {
	nowTime := time.Now().UnixMilli()
	s.sendBusinessNotification(DeviceBizNotification{
		DeviceID:  deviceID,
		BizType:   "offline",
		BizTime:   nowTime,
		BatchTime: nowTime,
		Iccid:     iccid,
		Data: map[string]string{
			"state": "OFF",
		},
		SensorAddr: "DTU",
	})
}

// 发送上线通知
func (s *TcpServer) sendOnlineNotification(deviceID string, iccid string) {
	nowTime := time.Now().UnixMilli()
	s.sendBusinessNotification(DeviceBizNotification{
		DeviceID:  deviceID,
		BizType:   "online",
		BizTime:   nowTime,
		BatchTime: nowTime,
		Iccid:     iccid,
		Data: map[string]string{
			"state": "ON",
		},
		SensorAddr: "DTU",
	})
}

/* ---------- TCP 服务器 ---------- */
type TcpServer struct {
	gnet.BuiltinEventEngine
	DeviceHandler      *transpond.DeviceHandler
	pool               *ants.Pool
	cfg                Config
	syncer             *DeviceStateSyncer
	businessNotifyChan chan DeviceBizNotification
	notificationDone   chan struct{}
	engine             *gnet.Engine
	shutdownChan       chan struct{}
	cmdCommandChan     chan dao.ChanDataCollectionCommand
	cmdCfgChan         chan dao.ChanDataCollectionConfig
	cmdLogic           *logic.CommandLogic
	startTime          time.Time
	manager            *connManager // 实例变量而非全局变量
}

// NewTcpServer 创建TCP服务器
func NewTcpServer(handler *transpond.DeviceHandler, cfg Config) *TcpServer {
	return &TcpServer{
		DeviceHandler:      handler,
		cfg:                cfg,
		notificationDone:   make(chan struct{}),
		shutdownChan:       make(chan struct{}),
		businessNotifyChan: make(chan DeviceBizNotification, cfg.NotificationBufferSize),
		manager:            newConnManager(cfg.MaxConnections), // 使用配置的最大连接数
	}
}

// sendBusinessNotification 发送业务通知 - 非阻塞发送带缓冲
func (s *TcpServer) sendBusinessNotification(notification DeviceBizNotification) {
	select {
	case s.businessNotifyChan <- notification:
		hlog.Debugf("设备 %s 业务通知已加入队列", notification.DeviceID)
	default:
		// 队列满时使用工作池异步等待，避免阻塞主流程
		if err := s.pool.Submit(func() {
			select {
			case s.businessNotifyChan <- notification:
				hlog.Debugf("设备 %s 业务通知延迟加入队列", notification.DeviceID)
			case <-time.After(1 * time.Second):
				hlog.Warnf("设备 %s 业务通知队列已满且等待超时，通知丢失", notification.DeviceID)
			case <-s.shutdownChan:
				hlog.Warnf("服务器关闭中，设备 %s 业务通知丢失", notification.DeviceID)
			}
		}); err != nil {
			hlog.Warnf("设备 %s 业务通知提交失败，线程池已满: %v", notification.DeviceID, err)
		}
	}
}

// handleBusinessNotifications 处理业务通知 - 批量处理优化
func (s *TcpServer) handleBusinessNotifications() {
	defer func() {
		if r := recover(); r != nil {
			hlog.Errorf("业务通知处理协程发生异常: %v", r)
		}
	}()

	// 批量处理缓冲区
	batch := make([]DeviceBizNotification, 0, 10)
	ticker := time.NewTicker(100 * time.Millisecond) // 100ms批量处理一次
	defer ticker.Stop()

	for {
		select {
		case notification := <-s.businessNotifyChan:
			batch = append(batch, notification)
			// 缓冲区满了立即处理
			if len(batch) >= cap(batch) {
				s.processNotificationBatch(batch)
				batch = batch[:0]
			}
		case <-ticker.C:
			// 定时处理剩余通知
			if len(batch) > 0 {
				s.processNotificationBatch(batch)
				batch = batch[:0]
			}
		case <-s.shutdownChan:
			// 处理剩余通知
			if len(batch) > 0 {
				s.processNotificationBatch(batch)
			}
			hlog.Info("业务通知处理协程收到关闭信号，退出")
			close(s.notificationDone)
			return
		}
	}
}

// 批量处理通知，减少频繁调用
func (s *TcpServer) processNotificationBatch(batch []DeviceBizNotification) {
	for _, notification := range batch {
		hlog.Debugf("处理设备 %s 业务通知", notification.DeviceID)
		s.DeviceHandler.OnMsg(&model.Device{ID: notification.DeviceID, Iccid: notification.Iccid}, notification.Data, notification.SensorAddr, notification.BizTime, notification.BatchTime)
	}
}

// OnBoot 服务器启动时调用
func (s *TcpServer) OnBoot(eng gnet.Engine) gnet.Action {
	s.engine = &eng
	s.startTime = time.Now()
	hlog.Info("TCP 服务器启动成功时间", s.startTime.Format(time.RFC3339))

	// 启动dchan处理协程
	go s.dchan()

	// 延迟启动设备状态同步
	go func() {
		hlog.Infof("将在 %v 后开始设备状态同步", s.cfg.StateSyncDelay)

		select {
		case <-time.After(s.cfg.StateSyncDelay):
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
			defer cancel()
			if err := s.syncer.SyncAllDevices(ctx); err != nil {
				hlog.Errorf("设备状态同步失败: %v", err)
			}
		case <-s.shutdownChan:
			hlog.Info("服务器正在关闭，取消设备状态同步")
			return
		}
	}()

	return gnet.None
}

// Shutdown 优雅关闭服务器
func (s *TcpServer) Shutdown() error {
	hlog.Info("开始优雅关闭服务器...")

	s.manager.setShuttingDown()
	close(s.shutdownChan)

	s.syncer.Stop()
	s.closeAllClients()

	if s.engine != nil {
		if err := s.engine.Stop(context.Background()); err != nil {
			hlog.Errorf("停止引擎失败: %v", err)
			return err
		}
	}

	select {
	case <-s.notificationDone:
		hlog.Info("通知处理协程已退出")
	case <-time.After(5 * time.Second):
		hlog.Warn("通知处理协程退出超时")
	}

	if s.pool != nil {
		hlog.Info("等待线程池完成剩余任务...")
		s.pool.ReleaseTimeout(s.cfg.ShutdownTimeout)
		hlog.Info("线程池已关闭")
	}

	hlog.Info("服务器已优雅关闭")
	return nil
}

// closeAllClients 关闭所有客户端连接 - 限制并发关闭数量
func (s *TcpServer) closeAllClients() {
	authClients := s.manager.getAllAuthClients()
	hlog.Infof("开始关闭 %d 个已认证客户端连接", len(authClients))

	// 限制并发关闭数量，避免资源竞争
	ch := make(chan struct{}, 10) // 最多10个并发关闭
	var wg sync.WaitGroup

	for _, client := range authClients {
		wg.Add(1)
		ch <- struct{}{}
		go func(c *Client) {
			defer wg.Done()
			defer func() { <-ch }()
			c.Send([]byte("server_shutdown\n"))
			_ = c.Close()
			c.Cleanup()
		}(client)
	}
	wg.Wait()

	tempClients := s.manager.getAllTempClients()
	hlog.Infof("开始关闭 %d 个未认证客户端连接", len(tempClients))

	for _, client := range tempClients {
		wg.Add(1)
		ch <- struct{}{}
		go func(c *Client) {
			defer wg.Done()
			defer func() { <-ch }()
			_ = c.Close()
			c.Cleanup()
		}(client)
	}
	wg.Wait()
}

// OnOpen 新连接时调用
func (s *TcpServer) OnOpen(c gnet.Conn) (out []byte, action gnet.Action) {
	remoteAddr := c.RemoteAddr().String()
	if s.manager.shuttingDown {
		hlog.Warnf("服务器正在关闭，拒绝新连接: %v", c.RemoteAddr())
		return nil, gnet.Close
	}

	// 检查IP是否被封禁
	if s.manager.IsIPBanned(remoteAddr) {
		hlog.Warnf("IP %s 已被封禁，拒绝新连接", remoteAddr)
		return nil, gnet.Close
	}

	// 检查连接数是否超过最大允许数
	if atomic.LoadInt32(&s.manager.totalConnections) >= s.manager.maxConnections {
		hlog.Warnf("当前连接数 %d 已超过最大允许数 %d，拒绝新连接: %v",
			atomic.LoadInt32(&s.manager.totalConnections), s.manager.maxConnections, c.RemoteAddr())
		return nil, gnet.Close
	}

	// 初始化客户端
	cli := NewClient(c, s.cfg.SendChanBufferSize)
	go s.startSender(cli)

	// 暂时不启动数据采集，等待认证后根据设备配置启动
	s.manager.temp.Store(cli.GetRemoteAddr(), cli)
	atomic.AddInt32(&s.manager.totalConnections, 1)
	hlog.Debugf("新客户端连接（未认证）：%v", cli.GetRemoteAddr())

	// 设置认证超时
	go func() {
		select {
		case <-time.After(s.cfg.AuthTimeout):
			if !cli.IsAuthenticated() {
				hlog.Warnf("客户端 %s 认证超时，关闭连接", cli.GetRemoteAddr())
				_ = cli.Close()
			}
		case <-cli.GetContext().Done():
			return
		}
	}()

	return
}

// startSender 启动消息发送协程 - 减少阻塞和日志
func (s *TcpServer) startSender(c *Client) {
	defer func() {
		close(c.GetSendDone())
		if r := recover(); r != nil {
			hlog.Errorf("客户端 %s 发送协程panic: %v", c.GetRemoteAddr(), r)
		}
	}()

	// 发送重试计数器，避免频繁日志
	sendFailCount := 0

	for {
		select {
		case data := <-c.GetSendChan():
			if err := c.SyncWrite(data); err != nil {
				sendFailCount++
				// 失败次数达到阈值才记录日志
				if sendFailCount%5 == 0 || sendFailCount == 1 {
					hlog.Warnf("向 %s 发送数据失败(第%d次): %v", c.GetRemoteAddr(), sendFailCount, err)
				}
				// 多次失败后关闭连接
				if sendFailCount >= 10 {
					hlog.Errorf("向 %s 发送数据多次失败，关闭连接", c.GetRemoteAddr())
					_ = c.Close()
					return
				}
			} else {
				sendFailCount = 0 // 重置计数器
			}
		case <-c.GetContext().Done():
			hlog.Debugf("客户端 %s 发送协程退出", c.GetRemoteAddr())
			return
		case <-s.shutdownChan:
			hlog.Debugf("服务器关闭，客户端 %s 发送协程退出", c.GetRemoteAddr())
			return
		}
	}
}

func (s *TcpServer) OnClose(c gnet.Conn, err error) gnet.Action {
	addr := c.RemoteAddr().String()
	atomic.AddInt32(&s.manager.totalConnections, -1)

	// 清理未认证连接
	if v, ok := s.manager.temp.LoadAndDelete(addr); ok {
		if client, ok := v.(*Client); ok {
			client.Cleanup()
			hlog.Debugf("未认证客户端断开：%v, 错误: %v", addr, err)
		}
		return gnet.None
	}

	// 清理已认证连接
	if v, ok := s.manager.auth.Load(addr); ok {
		if client, ok := v.(*Client); ok {
			client.SetLastOfflineTime(time.Now())
			s.manager.removeAuth(client)

			device := client.GetDevice()
			if device != nil {
				s.deviceOffline(device)
			}
			client.Cleanup()
			hlog.Debugf("已认证客户端断开：%v, 错误: %v", addr, err)
		}
	}

	return gnet.None
}

// OnTraffic 消息处理 - 减少内存分配和日志输出
func (s *TcpServer) OnTraffic(c gnet.Conn) (action gnet.Action) {
	if s.manager.shuttingDown {
		return gnet.Close
	}

	clientAddr := c.RemoteAddr()
	addrStr := clientAddr.String()

	// 读取数据
	data, err := c.Next(-1)
	if err != nil {
		hlog.Warnf("客户端 %s 读取数据错误: %v", addrStr, err)
		return gnet.Close
	}

	// 过滤空数据
	if len(data) == 0 {
		return gnet.None
	}

	// 减少Debug日志输出频率 - 每10条记录一次
	if atomic.AddInt32(&debugLogCounter, 1)%10 == 0 {
		hlog.Debugf("收到客户端 %s 数据: %s", addrStr, string(data))
	}

	// 检查非法数据并封禁IP
	if s.isIllegalData(string(data), addrStr) {
		hlog.Warnf("客户端 %s 发送非法数据，已封禁IP 24小时", addrStr)
		s.manager.BanIP(addrStr, 24*time.Hour)
		return gnet.Close
	}

	// 处理未认证客户端
	if tempVal, ok := s.manager.temp.Load(addrStr); ok {
		if cli, ok := tempVal.(*Client); ok {
			cli.SetIccid(string(data))
			// 使用非阻塞方式提交任务，避免线程池满时阻塞
			if err := s.pool.Submit(func() {
				s.handleAuth(cli)
			}); err != nil {
				hlog.Errorf("客户端 %s 提交认证任务失败(线程池满): %v", addrStr, err)
				_ = cli.Close()
			}
			return gnet.None
		}
	}

	// 处理已认证客户端
	if authVal, ok := s.manager.auth.Load(addrStr); ok {
		if cli, ok := authVal.(*Client); ok {
			// 处理心跳响应
			if string(data) == "pong" {
				s.handlePong(cli)
				return gnet.None
			}

			// 减少日志输出
			//if atomic.AddInt32(&modbusLogCounter, 1)%20 == 0 {
			hlog.Debugf("收到客户端 %s 数据: %s", addrStr, hex.EncodeToString(data))
			//}

			// 使用非阻塞方式提交任务
			if err := s.pool.Submit(func() {
				s.handleModbusData(cli, data)
			}); err != nil {
				hlog.Errorf("客户端 %s 提交数据处理任务失败(线程池满): %v", addrStr, err)
			}
		}
	}

	return gnet.None
}

// 全局计数器，用于控制日志输出频率
var (
	debugLogCounter  int32
	modbusLogCounter int32
)

// handlePong 处理心跳响应 - 减少日志
func (s *TcpServer) handlePong(c *Client) {
	now := time.Now()
	delay := now.Sub(c.GetLastPingSent())
	if delay > s.cfg.PingResponseTimeout/2 {
		// 减少警告日志频率
		if atomic.AddInt32(&heartbeatWarnCounter, 1)%5 == 0 {
			hlog.Warnf("设备 %s 心跳响应延迟较高: %v", c.GetDeviceID(), delay)
		}
	}

	c.SetLastPong(now)
	c.ResetMissedPings()
	// 仅在调试模式输出日志
	hlog.Debugf("收到设备 %s 的心跳响应", c.GetDeviceID())
}

// 全局计数器，用于控制心跳日志输出频率
var (
	heartbeatWarnCounter int32
)

/* ---------- 认证处理 ---------- */
func (s *TcpServer) handleAuth(cli *Client) {
	if s.manager.shuttingDown {
		hlog.Info("服务器正在关闭，拒绝新认证")
		_ = cli.Close()
		return
	}

	select {
	case <-cli.GetContext().Done():
		hlog.Debugf("客户端 %s 已断开，终止认证", cli.GetRemoteAddr())
		return
	default:
	}

	iccid := cli.GetIccid()

	device, ok, err := s.queryDeviceAuth(cli.GetContext(), iccid)
	if err != nil {
		hlog.Errorf("查询设备 %s 失败: %v", iccid, err)
		_ = cli.Close()
		return
	}

	if !ok {
		hlog.Infof("设备 %s 未注册或禁用，关闭连接", iccid)
		_ = cli.Close()
		return
	}

	cli.SetDeviceID(device.Iccid)
	cli.SetDevice(device)
	cli.SetProtocolType(device.ProtocolType)
	s.manager.moveToAuth(cli)

	// 优化定时器使用
	if timer := cli.GetPingTimer(); timer != nil {
		timer.Stop()
	}
	// 使用固定间隔的定时器，避免频繁调整
	timer := time.AfterFunc(s.cfg.PingInterval, func() {
		s.checkHeartbeat(cli)
	})
	cli.SetPingTimer(timer)

	// 提交配置刷新任务
	if err := s.pool.Submit(func() {
		s.RefreshDeviceConfig(device.ID, iccid)
	}); err != nil {
		hlog.Errorf("提交设备 %s 配置刷新任务失败: %v", iccid, err)
	}

	//发送上线通知
	s.sendOnlineNotification(device.ID, iccid)
}

/* ---------- 心跳检测 ---------- */
func (s *TcpServer) checkHeartbeat(c *Client) {
	select {
	case <-c.GetContext().Done():
		return
	case <-s.shutdownChan:
		return
	default:
	}

	// 检查心跳丢失是否超限
	if c.GetMissedPings() >= s.cfg.MaxMissedPings {
		hlog.Warnf("设备 %s 连续丢失 %d 次心跳，断开连接", c.GetDeviceID(), c.GetMissedPings())
		_ = c.Close()
		return
	}

	// 发送Ping
	if err := c.Send([]byte("ping\n")); err != nil {
		hlog.Warnf("向设备 %s 发送心跳失败: %v", c.GetDeviceID(), err)
		_ = c.Close()
		return
	}
	c.SetLastPingSent(time.Now())
	c.IncrementMissedPings() // 预先计数，等待响应

	// 固定间隔重试
	if timer := c.GetPingTimer(); timer != nil {
		timer.Reset(s.cfg.PingInterval)
	}
}

/* ---------- 设备管理 ---------- */
func (s *TcpServer) queryDeviceAuth(ctx context.Context, iccid string) (*model.Device, bool, error) {
	deviceLogic := logic.NewDeviceLogic(ctx)
	device, err := deviceLogic.FindByIccid(iccid)
	if err != nil {
		return nil, false, fmt.Errorf("查询设备失败: %w", err)
	}

	if device == nil || device.Enabled != 1 {
		return nil, false, nil
	}

	device.Online = 1
	device.OfflineReason = "ON"
	device.LastTime = time.Now()
	if err := deviceLogic.UpdateDevice(device); err != nil {
		return nil, false, fmt.Errorf("更新设备状态失败: %w", err)
	}
	return device, true, nil
}

func (s *TcpServer) deviceOffline(d *model.Device) {
	if err := s.pool.Submit(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		device := query.Device
		result, err := device.WithContext(ctx).
			Where(device.ID.Eq(d.ID)).
			Updates(map[string]interface{}{
				"online":         0,
				"last_time":      time.Now(),
				"offline_reason": "normal",
			})

		if err != nil {
			hlog.Errorf("更新设备 %s 离线状态失败: %v", d.ID, err)
			return
		}

		if result.RowsAffected == 0 {
			hlog.Warnf("设备 %s 记录不存在，无法更新离线状态", d.ID)
		}
		//下线通知
		s.sendOfflineNotification(d.ID, d.Iccid)
	}); err != nil {
		hlog.Errorf("提交设备下线任务失败: %v", err)
	}
}

/* ---------- 发送消息接口 ---------- */
func (s *TcpServer) SendMsg(clientId string, data []byte) error {
	if s.manager.shuttingDown {
		return errors.New("服务器正在关闭，无法发送消息")
	}

	if v, ok := s.manager.auth.Load(clientId); ok {
		if cli, ok := v.(*Client); ok {
			return cli.Send(data)
		}
	}
	return errors.New("设备未在线或未认证")
}

func (s *TcpServer) SendMsgByDeviceID(deviceID string, data []byte) error {
	if s.manager.shuttingDown {
		return errors.New("服务器正在关闭，无法发送消息")
	}

	client, ok := s.manager.getClientByDeviceID(deviceID)
	if !ok {
		return errors.New("设备未在线")
	}
	return client.Send(data)
}

// UnregisterConnection 注销设备连接
func (s *TcpServer) UnregisterConnection(deviceID int64) {
	// 实现保持不变
}

// UpdateDeviceCommand 向设备推送单条指令更新
func (s *TcpServer) UpdateDeviceCommand(deviceID string, cmd dao.DataCollectionCommand) error {
	client, ok := s.manager.getClientByDeviceID(deviceID)
	if !ok {
		return errors.New("设备未在线")
	}
	client.UpdateCommandConfig(cmd)
	return nil
}

// SetDeviceCommands 向设备推送完整指令配置
func (s *TcpServer) SetDeviceCommands(deviceID string, cfg dao.DataCollectionConfig) error {
	client, ok := s.manager.getClientByDeviceID(deviceID)
	if !ok {
		return errors.New("设备未在线")
	}
	client.ConfigureDataCollection(cfg)
	return nil
}

// 刷新配置 - 增加缓存和错误处理
func (s *TcpServer) RefreshDeviceConfig(deviceID string, iccid string) {
	// 检查设备是否仍在线
	client, ok := s.manager.getClientByDeviceID(iccid)
	if !ok {
		hlog.Debugf("设备 %s 已离线，无需刷新配置", iccid)
		return
	}

	// 从数据库获取设备的指令配置并应用
	cmdConfig, err := s.cmdLogic.GetDeviceCollectionConfig(context.Background(), deviceID)
	if err != nil {
		hlog.Warnf("获取设备 %s 指令配置失败: %v，使用默认配置", iccid, err)
		return
	}

	deviceLogic := logic.NewDeviceLogic(context.Background())
	device, err := deviceLogic.FindByIccid(iccid)
	if err != nil {
		hlog.Warnf("获取设备 %s 信息失败: %v", iccid, err)
		return
	}

	if device != nil {
		cmdConfig.Interval = time.Duration(device.CollectionInterval) * time.Second
		client.SetProtocolType(device.ProtocolType)
		client.ConfigureDataCollection(cmdConfig)
		s.DeviceHandler.RefreshDeviceConfig(deviceID, iccid)
		hlog.Debugf("刷新设备 %s 配置", iccid)
	}
}

// dchan 处理通道数据 - 优化流量控制
func (s *TcpServer) dchan() {
	defer hlog.Info("dchan处理协程已退出")

	for {
		select {
		case <-s.shutdownChan:
			return
		case cfg, ok := <-dchan.CfgChan:
			if !ok {
				hlog.Warn("CfgChan已关闭")
				return
			}
			if err := s.pool.Submit(func() {
				s.RefreshDeviceConfig(cfg.DeviceId, cfg.Iccid)
			}); err != nil {
				hlog.Errorf("提交配置刷新任务失败: %v", err)
			}
		case cmd, ok := <-dchan.DeviceSendChan:
			if !ok {
				hlog.Warn("DeviceSendChan已关闭")
				return
			}
			if err := s.SendMsgByDeviceID(cmd.Iccid, cmd.Cmd); err != nil {
				hlog.Errorf("发送指令 %s 失败: %v", cmd.Cmd, err)
			}
		}
	}
}

// 减少插件调用频率和错误处理
func (s *TcpServer) handleModbusData(cli *Client, data []byte) {
	protocolType := cli.protocolType
	if protocolType == "" {
		hlog.Warnf("设备 %s 协议类型未设置", cli.GetDeviceID())
		return
	}

	// 从插件管理器获取协议插件
	protocolManager, err := plugin_manager.GetGlobalManager()
	if err != nil {
		hlog.Errorf("获取全局插件管理器失败: %v", err)
		return
	}

	// 获取协议插件
	protocol, ok := protocolManager.GetPlugin(protocolType)
	if !ok {
		hlog.Errorf("协议 %s 不存在", protocolType)
		return
	}

	// 解析数据
	response, err := protocol.ParseResponse(context.Background(), &plugin.ParseResponseRequest{
		Data: data,
	})
	if err != nil {
		hlog.Errorf("协议 %s 解析数据失败: %v", protocolType, err)
		return
	}

	if response == nil {
		hlog.Debugf("协议 %s 未解析到有效数据", protocolType)
		return
	}

	// 减少日志输出
	//if atomic.AddInt32(&parsedDataLogCounter, 1)%50 == 0 {
	hlog.Debugf("协议 %s 解析后的数据: %v", protocolType, response)
	//}

	device := cli.GetDevice()
	if device != nil {
		s.sendBusinessNotification(DeviceBizNotification{
			DeviceID:   device.Iccid,
			Iccid:      device.Iccid,
			BizTime:    time.Now().UnixMilli(),
			BatchTime:  cli.GetDataCollectionStartTime(),
			BizType:    "message",
			Data:       response,
			SensorAddr: response.GetData().GetProtocol(),
		})
	}
}

// 全局计数器，用于控制解析数据日志输出频率
var parsedDataLogCounter int32

/* ---------- 服务器启动 ---------- */
func Start(ctx context.Context, handler *transpond.DeviceHandler, cfg Config) error {
	// 调整工作池参数，避免创建过多goroutine
	pool, err := ants.NewPool(cfg.WorkerPoolSize,
		ants.WithPreAlloc(true),
		ants.WithMaxBlockingTasks(500), // 增加阻塞队列大小
		ants.WithNonblocking(false),    // 允许适当阻塞，避免任务丢失
	)
	if err != nil {
		return fmt.Errorf("初始化线程池失败: %w", err)
	}
	defer pool.Release()

	server := NewTcpServer(handler, cfg)
	server.pool = pool
	server.syncer = NewDeviceStateSyncer(server, pool, cfg)

	hlog.Infof("启动 TCP 服务器，监听地址: %s", cfg.ListenAddr)
	// 初始化指令仓库和逻辑
	cmdRepo := repo.NewCommandRepo(query.Q)
	deviceRepo := repo.NewDeviceRepo(query.Q)
	server.cmdLogic = logic.NewCommandLogic(cmdRepo, deviceRepo)

	eg, egCtx := errgroup.WithContext(ctx)

	//启动业务通知处理
	eg.Go(func() error {
		server.handleBusinessNotifications()
		return nil
	})

	// 监听系统信号
	eg.Go(func() error {
		sigChan := make(chan os.Signal, 1)
		signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)

		select {
		case sig := <-sigChan:
			hlog.Infof("收到信号 %s，开始优雅关闭", sig.String())
			return server.Shutdown()
		case <-egCtx.Done():
			hlog.Info("上下文已取消，启动关闭流程")
			return server.Shutdown()
		}
	})

	// 启动gnet服务器
	eg.Go(func() error {
		// 自定义事件循环线程数，避免过度并行导致的CPU上下文切换

		return gnet.Run(server, cfg.ListenAddr,
			// 关闭多核心模式，改用手动指定工作线程数，减少锁竞争
			gnet.WithMulticore(true),

			// 仅在确实需要gnet内置定时器时开启
			gnet.WithTicker(false),

			// 关闭ReusePort，某些场景下会导致额外的CPU开销
			gnet.WithReusePort(false),

			// 调整TCP保活参数，避免频繁保活探测
			gnet.WithTCPKeepAlive(15*time.Minute),

			// 合理设置缓冲区大小，避免频繁内存分配
			gnet.WithReadBufferCap(8192),
			gnet.WithWriteBufferCap(8192),
		)
	})

	if err := eg.Wait(); err != nil {
		hlog.Errorf("服务器运行错误: %v", err)
		return err
	}

	hlog.Info("服务器已退出")
	return nil
}
